package com.mallcai.bigdata.ladon.datasource;

import com.cake.bigdata.starter.exception.category.BusinessException;
import com.mallcai.bigdata.ladon.dpl.entity.common.Datasource;
import com.mallcai.bigdata.ladon.dpl.vo.DatasourceType;
import com.mallcai.bigdata.ladon.vo.DataVo;
import com.mallcai.bigdata.ladon.vo.QueryParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.xpack.sql.jdbc.EsDataSource;
import tk.mybatis.mapper.util.Assert;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Map;
import java.util.Properties;

/**
 * Created by oneape<oneape15@163.com>
 * Created 2019-06-04 10:18.
 * Modify:
 */
@Slf4j
public class ElasticsearchDataSource extends AbstractDataSource {

    public ElasticsearchDataSource(Datasource ds) {
        super(ds);
    }

    @Override
    public DataSource initDataSource(Datasource ds) {
        Assert.isTrue(ds != null, "数据源信息为空");
        if (StringUtils.isBlank(ds.getJdbcUrl()) ||
                StringUtils.startsWith("jdbc:es://", StringUtils.lowerCase(ds.getJdbcUrl()))) {
            throw new RuntimeException("不合法的地址~");
        }

        try {
            Class.forName(DatasourceType.ElasticSearch.getDriver()).newInstance();
        } catch (Exception e) {
            throw new BusinessException("驱动包不存在", e);
        }

        Properties connProps = new Properties();

        // 访问加密的ES
        if (StringUtils.isNotBlank(ds.getUsername())) {
            connProps.put("user", StringUtils.trim(ds.getUsername()));
        }
        if (StringUtils.isNotBlank(ds.getPassword())) {
            connProps.put("password", StringUtils.trim(ds.getPassword()));
        }

        EsDataSource dataSource = new EsDataSource();
        dataSource.setUrl(ds.getJdbcUrl());
        dataSource.setProperties(connProps);

        return dataSource;
    }

    /**
     * 获取连接对象.
     *
     * @return Connection
     */
    @Override
    public Connection getConnection() throws BusinessException {
        try {
            return dataSource.getConnection();
        } catch (Exception e) {
            log.error("获取数据源连接失败", e);
            throw new BusinessException("获取数据源连接失败", e);
        }
    }

    /**
     * 是否支持分页
     *
     * @return boolean
     */
    @Override
    public boolean supportPage() {
        return true;
    }

    /**
     * 执行查询语句
     *
     * @param queryParam QueryParam
     * @return DataVo
     */
    @Override
    public DataVo execSql(QueryParam queryParam) throws BusinessException {
        String sql = queryParam.getRawSql();
        if (StringUtils.isBlank(sql)) {
            throw new BusinessException("查询SQL为空");
        }

        //判断在sql的最后面是否有LIMIT存在
        boolean isExistLimit = isEndWithLimit(sql);
        int limitSize = ONE_MAX_SIZE;
        if (queryParam.getLimitSize() != null) {
            limitSize = queryParam.getLimitSize();
        }
        if (isExistLimit) {
            sql = editLimitIfOverMax(sql);
        }
        if (!isExistLimit && queryParam.isNeedLimit()) {
            sql = sql + " " + KEY_LIMIT + limitSize;
        }

        Connection conn = getConnection();
        Map<Integer, Object> params = queryParam.getParams();

        PreparedStatement preStmt = null;
        ResultSet resultSet = null;
        try {
            preStmt = conn.prepareStatement(sql);
            //写入预编译时的参数
            if (params != null && params.size() > 0) {
                Object obj;
                String objStr;
                for (Map.Entry<Integer, Object> entry : params.entrySet()) {
                    obj = entry.getValue();
                    Integer key = entry.getKey();
                    objStr = String.valueOf(obj);
                    if (obj instanceof java.lang.String) { //字符串
                        preStmt.setString(key, objStr);
                    } else if (obj instanceof java.lang.Integer) { //整数类型
                        preStmt.setInt(key, Integer.parseInt(objStr));
                    } else if (obj instanceof java.lang.Boolean) { //布尔类型
                        preStmt.setBoolean(key, Boolean.valueOf(objStr));
                    } else if (obj instanceof java.lang.Double) {
                        preStmt.setDouble(key, Double.valueOf(objStr));
                    } else {
                        preStmt.setObject(key, obj);
                    }
                }
            }

            log.debug("开始执行sql!!");
            //执行sql
            long startTime = System.nanoTime();
            resultSet = preStmt.executeQuery();
            long endTime = System.nanoTime();
            long waitTime = (endTime - startTime) / NS_MS;  //将纳秒 转换成 毫秒
            //超过800ms的记录sql
            if (waitTime > 800) {
                log.warn("耗时:{}ms, 执行sql:{}", waitTime, preStmt.toString());
            }

            DataVoExtractor dataVoExtractor = new DataVoExtractor(queryParam.getDataTypes());
            return dataVoExtractor.extractData(resultSet);
        } catch (Exception e) {
            log.error("查询运行错误: {}", sql, e);
            throw new BusinessException("查询运行错误", e);
        } finally {
            try {
                if (resultSet != null) {
                    resultSet.close();
                }
            } catch (Exception e) {
                log.error("关闭resultSet失败! msg:{}", e.getMessage());
            }
            try {
                if (preStmt != null) {
                    preStmt.close();
                }
            } catch (Exception e) {
                log.error("关闭stmt失败! msg:{}", e.getMessage());
            }
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (Exception e) {
                log.error("关闭Connection失败! msg:{}", e.getMessage());
            }
        }
    }

    private String formatSql(QueryParam queryParam) {
        String tmpSql = queryParam.getRawSql();

        int index = 1;
        Map<Integer, Object> params = queryParam.getParams();
        while (StringUtils.indexOf(tmpSql, "?") > -1) {
            Object obj = params.get(index);

            String value = "";
            if (obj instanceof String) {
                value = "'" + String.valueOf(obj) + "'";
            } else {
                value = String.valueOf(obj);
            }

            // 每次只替换第一个;
            tmpSql = StringUtils.replace(tmpSql, "?", value, 1);
        }

        // 限制返回条数
        long limitSize = 1000;
        if (queryParam.getLimitSize() != null && queryParam.getLimitSize() > 0) {
            limitSize = queryParam.getLimitSize();
        }
        tmpSql += " LIMIT " + limitSize;
        return tmpSql;
    }
}
